package com.chencong.online.function;

import com.chencong.online.bean.UniqueVisitorBean;
import com.chencong.online.bean.UserBehaviorBean;
import com.chencong.online.utils.TimestampTransformUtil;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;
import java.util.HashSet;

/**
 * @program: user-behavior-analysis-online
 * @ClassName UVCountAllWindowFunc
 * @description:获取窗口内的数据，利用set去重
 * @author: chencong
 * @create: 2021-12-28 15:31
 **/
public class UVCountAllWindowFunc implements AllWindowFunction<UserBehaviorBean, UniqueVisitorBean, TimeWindow> {

    @Override
    public void apply(TimeWindow window, Iterable<UserBehaviorBean> values, Collector<UniqueVisitorBean> out) throws Exception {
        //获取hashset
        HashSet<Long> set = new HashSet<>();
        long end = window.getEnd();
        LocalDateTime endTime = TimestampTransformUtil.milliTimestampToLocalDateTime(end);
        //循环把数据放入set进行去重
        for (UserBehaviorBean value : values) {
            Long userId = value.getUserId();
            set.add(userId);
        }
        out.collect(new UniqueVisitorBean(endTime.toString(), (long) set.size()));
    }
}
